package com.grj.payment.consumer;

/**
 * @author ganru
 * @date 2025/10/19 5:19
 */

import com.grj.common.model.Order;
import com.grj.common.model.ProcessedEvent;
import com.grj.payment.repository.ProcessedEventRepository;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;

@Component
public class PaymentConsumer {
    private final KafkaTemplate<String, Object> kafkaTemplate;
    private final ProcessedEventRepository processedEventRepository;
    public PaymentConsumer(KafkaTemplate<String, Object> kafkaTemplate, ProcessedEventRepository processedEventRepository) {
        this.kafkaTemplate = kafkaTemplate;
        this.processedEventRepository = processedEventRepository;
    }
    @KafkaListener(topics = "order-created", groupId = "payment-service")
    public void handleOrderCreated(Order order) {
        String eventId = order.getOrderId() + ":order-created";
        String consumerName = "payment-service";

        // 幂等检查
        if (processedEventRepository.findByEventIdAndConsumer(eventId, consumerName).isPresent()) {
            System.out.println("已处理，跳过 " + eventId);
            return;
        }

        // 处理逻辑（假设要写 DB，请包在事务中）
        try {
            // do business: 写支付表/调用支付网关/更新订单状态（若更新订单 DB, 要通过 API 或 DB 事务）
            // 处理完成后，保存 processed_event 记录以保证幂等
            ProcessedEvent p = new ProcessedEvent();
            p.setEventId(eventId);
            p.setConsumer(consumerName);
            processedEventRepository.save(p);
        } catch (Exception e) {
            throw e; // 抛出异常触发重试或 DLQ
        }
    }
}
